﻿//Description:同步表全量数据
//Copyright (c) : 通力凯顿（北京）系统集成有限公司
//Writer:Wangjh
//create Date:2023-8-9
//Rewriter:
//Rewrite Date:
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using BusinessRule.Common;
using System.Data;
using DatabaseAccess.DBUtility;
using System.Configuration;
using BusinessRule.DTO;
using System.IO;
using System.Collections;
using Newtonsoft.Json;
using System.Threading.Tasks;
using System.Reflection;
using Newtonsoft.Json.Linq;

namespace OracleTableSync
{
    public class SyncTableAllData
    { 
        private static readonly String receivePath = ConfigurationManager.AppSettings["ReceivePath"].ToString();//接收文件路径
        private static readonly String receiveFileBak = ConfigurationManager.AppSettings["receiveFileBak"].ToString();//接收文件路径-历史
        private static readonly String receiveFileBakError = ConfigurationManager.AppSettings["receiveFileBakError"].ToString();//接收文件路径-历史，处理错误路径
        private static readonly String orgDb = ConfigurationManager.AppSettings["orgDb"].ToString();
        private static readonly String toDb = ConfigurationManager.AppSettings["toDb"].ToString();
        private static readonly String tabMapFilePath = ConfigurationManager.AppSettings["tabMapFilePath"].ToString();
        private static List<JObject> tabMapTab;//字段映射关系
        private static readonly int maxParallel = Environment.ProcessorCount;//最大并行数
        private const int maxData = 1000;//一个json文件容纳的最大数据量，暂定2000

        /// <summary>
        /// 执行入口
        /// </summary>
        public void Excute()
        {
            LogHelp.SetConfig();
            //处理推送
            DealSyncAllData();
            //处理接收
            ReceiveSyncAllData();
        }

        #region 处理全量推送数据，经本地测试9分钟能推送100万数据量
        /// <summary>
        /// 获取需要全量同步的表
        /// </summary>
        /// <returns></returns>
        DataTable GetSyncAllDataTable()
        {
            DataTable dt = DatabaseHelper.Query_ReturnDatTable(@"select stt.id TableTaskID,stt.operatetype,
st.id,st.tablename,st.uniquecolname,stt.Createtime from sync_tableinfo st 
INNER JOIN Sync_TableTask stt on stt.tablename=st.tablename and stt.ishandle=0
where stt.OperateType =6
order by stt.lastchangetime ");
            return dt;
        }

         /// <summary>
        /// 处理需要全量同步的数据
        /// </summary>
        public void DealSyncAllData()
        {
            DataTable dt = GetSyncAllDataTable();

            foreach (DataRow dr in dt.Rows)
            {
                try
                {
                    DataTable dtColInfo = CommonHelper.getTableColInfo(dr["tableName"].ToString());
                    DealSyncAllDataEach(dr, dtColInfo);
                }
                catch (Exception ex)
                {
                    LogHelp.WriteError(ex.Message, ex);
                }
            }

        }

        /// <summary>
        /// 处理每页的数据
        /// </summary>
        /// <param name="dr"></param>
        /// <param name="dtColInfo"></param>
        void DealSyncAllDataEachPage(DataRow dr, DataTable dtColInfo, long i, int allCount, String createTabSql)
        {
            List<Dictionary<String, Object>> jsonFile = new List<Dictionary<String, Object>>();
            String pushID = Guid.NewGuid().ToString();

            uMESPagingDataDTO paraPage = new uMESPagingDataDTO();
            paraPage.CurrentPageIndex = Convert.ToInt32(i );
            paraPage.PageSize = Convert.ToInt32(maxData);
            paraPage.strSQL = CommonHelper.generateSelectSql(dtColInfo, dr["tablename"].ToString());
            DataTable dt = DatabaseHelper.RunProcedureForPage(paraPage).DBTable;


            //存入json文件
            Dictionary<String, Object> eachJson = new Dictionary<String, Object>();
            eachJson.Add("type", "syncAllData");
            eachJson.Add("orgDb", orgDb);
            eachJson.Add("toDb", toDb);
            eachJson.Add("pushID", pushID);
            eachJson.Add("tableTaskID", dr["TableTaskID"].ToString());
            eachJson.Add("insertData", CommonHelper.generateStrJSON(dt, dtColInfo));
            eachJson.Add("tableName", dr["TableName"].ToString());
            eachJson.Add("uniqueColName", dr["UniqueColName"].ToString());
            eachJson.Add("operateType", dr["operateType"].ToString());
            eachJson.Add("allCount", allCount);
            eachJson.Add("pageSize", paraPage.PageSize);
            eachJson.Add("currentPageIndex", paraPage.CurrentPageIndex);
            if (paraPage.CurrentPageIndex == 1)
            { //第一页增加创建表结构的语句
                String _createTabSql = DatabaseHelper.GetSingle(String.Format(createTabSql, dr["TableName"].ToString())).ToString();
                //eachJson.Add("createTabSql", _createTabSql);
            }


            jsonFile.Add(eachJson);
            String prefix = "FullData_";
            //if (i == 1) {
            //    prefix += "1_";
            //}
            prefix += i.ToString()+"_";
            String fileName = CommonHelper.writeJsonToFile(jsonFile, prefix + jsonFile[0]["tableName"].ToString());
            //CommonHelper.savePushFileRecord(jsonFile, fileName);//存入文件推送记录

        }

         /// <summary>
        /// 处理每张表结构
        /// </summary>
        /// <param name="dr"></param>
        void DealSyncAllDataEach(DataRow dr, DataTable dtColInfo)
        {
            String sourceIP = CommonHelper.GetIP();
            List<ExcuteEntity> exs = new List<ExcuteEntity>();
            String createTabSql = "select DBMS_METADATA.GET_DDL('TABLE',upper( '{0}')) ddlsql from dual";
            int allCount = int.Parse(DatabaseHelper.GetSingle("select count(1) from " + dr["tablename"].ToString() + "").ToString());

            double pageCount = Math.Ceiling((double)allCount / maxData);//获取总页数
            //pageCount = 500;//临时用

             var  paraOptions = new ParallelOptions();
             paraOptions.MaxDegreeOfParallelism = maxParallel;

            //必须先生成第一页，后续的可以并行循环
            DealSyncAllDataEachPage(dr, dtColInfo, 1, allCount, createTabSql);

            Parallel.For(2, (long)pageCount+1, paraOptions, (i) =>
            {//一页生成一个json文件
                DealSyncAllDataEachPage(dr, dtColInfo, i, allCount, createTabSql);
            });

            //处理完毕后更新状态
            exs.Add(new ExcuteEntity()
            {
                TableName = "Sync_TableTask",
                ExcuteType = ExcuteEntity.ExcuteTypeEnum.update,
                ExcuteFileds = new List<FieldEntity>() { new FieldEntity("IsHandle", 1, FieldEntity.FieldTypeEnum.Numer), 
                new FieldEntity("LastChangeTime",DateTime.Now, FieldEntity.FieldTypeEnum.Timestamp)},
                WhereFileds = new List<FieldEntity>() { new FieldEntity("Id", dr["TableTaskID"].ToString(), FieldEntity.FieldTypeEnum.Str) }
            });
            DatabaseHelper.ExecuteDataByEntitys(exs);


        }
            
        #endregion


        #region 处理全量接收数据,经本地测试15分钟能处理100万数据量
        /// <summary>
        /// 处理接收到的全量表数据
        /// </summary>
        public void ReceiveSyncAllData()
        { 
            using (StreamReader streamReader = new StreamReader(tabMapFilePath))
            {
                string jsonText = streamReader.ReadToEnd();
                 tabMapTab = JsonConvert.DeserializeObject<List<JObject>>(jsonText); 
            }

            DirectoryInfo mydir = new DirectoryInfo(receivePath);
            FileInfo[] fis = mydir.GetFiles();
            //先处理全量同步的第一页数据
            FileInfo[] firstFils = fis.Where((item) => {
                return item.Name.Contains("_FullData_1_");
            }).ToArray();
            foreach (FileInfo fi in firstFils)
            {
                try
                {
                    DealSyncDataEachFile(fi,true);
                }
                catch (Exception ex)
                {
                    LogHelp.WriteError(ex.Message, ex);
                    //处理异常
                    CommonHelper.moveFile(fi, receiveFileBakError);
                }
            }
            //并行处理第二页到最后一页的数据
            FileInfo[] otherFils = fis.Where((item) =>
            {
                return item.Name.Contains("_FullData_") && !item.Name.Contains("_FullData_1_");
            }).ToArray();
            var paraOptions = new ParallelOptions();
            paraOptions.MaxDegreeOfParallelism = maxParallel;

            Parallel.ForEach(otherFils, paraOptions, (fi) =>
            {
                try
                {
                    DealSyncDataEachFile(fi,false);
                }
                catch (Exception ex)
                {
                    LogHelp.WriteError(ex.Message, ex);
                    //处理异常
                    CommonHelper.moveFile(fi, receiveFileBakError);
                }
            });
             

        }

        /// <summary>
        /// 处理单个json文件，全量数据同步
        /// </summary>
        /// <param name="fi"></param>
        void DealSyncDataEachFile(FileInfo fi, Boolean isFirstPage)
        {
            if (fi.Extension != ".json")
            {
                return;
            }
            List<ExcuteEntity> exs = new List<ExcuteEntity>();
            ArrayList sqlList = new ArrayList();
            String receiveID = Guid.NewGuid().ToString();//一次接收的id 

            using (StreamReader streamReader = new StreamReader(fi.FullName))
            {
                string jsonText = streamReader.ReadToEnd();

                List<Dictionary<String, Object>> obj = JsonConvert.DeserializeObject<List<Dictionary<String, Object>>>(jsonText);

                Parallel.ForEach(obj, (eachData) =>
                {
                    if (eachData["type"].ToString() != "syncAllData")
                        return;

                    String tableName = eachData["tableName"].ToString();
                    String orgDb = eachData["orgDb"].ToString();
                    String toDb = eachData["toDb"].ToString();
                    //找到映射关系
                    JObject tab = CommonHelper.getMapTabByTableName(tabMapTab, tableName,orgDb);
                    if (tab == null) {
                        LogHelp.WriteLog("数据同步-未找到表：" + tableName + "的映射信息，,文件："+fi.Name);
                        return;
                    }

                    if (isFirstPage) {
                        sqlList.Add("TRUNCATE TABLE " + eachData["tableName"].ToString());
                    }
                    //if (eachData["currentPageIndex"].ToString() == "1" && eachData.ContainsKey("createTabSql"))
                    //{ //第一页数据，需要drop表再重新创建表
                    //    if (DatabaseHelper.GetSingle("SELECT COUNT(1) FROM USER_TABLES WHERE TABLE_NAME ='" + eachData["tableName"].ToString() + "'").ToString() == "1")
                    //    { //已存在表结构
                    //        sqlList.Add("DROP TABLE " + eachData["tableName"].ToString());
                    //    }
                    //    sqlList.Add(eachData["createTabSql"]);
                    //}
                    //执行语句
                    exs.AddRange(GetExcuteData(eachData,tab,orgDb,toDb));
                    //
                });
                 
                //CommonHelper.saveReceiveFileRecord(obj, fi.Name);//存入文件接收记录
            } 
            if (sqlList.Count > 0)
                DatabaseHelper.ExecuteSqlListTran(sqlList);
            if (exs.Count>0)
                DatabaseHelper.ExecuteDataByEntitys(exs);
            //移动文件
            CommonHelper.moveFile(fi, receiveFileBak);

        }

        /// <summary>
        /// 将每行数据解析成entity执行对象
        /// </summary>
        /// <param name="eachData"></param>
        /// <returns></returns>
        List<ExcuteEntity> GetExcuteData(Dictionary<String, Object> eachData, JObject thisTab, String orgDb, String toDb)
        {
            List<ExcuteEntity> result = new List<ExcuteEntity>();
            List<Dictionary<String, Object>> insertData = JsonConvert.DeserializeObject<List<Dictionary<String, Object>>>(eachData["insertData"].ToString());
            if (eachData["operateType"].ToString() == "6")
            {
                foreach (Dictionary<String, Object> insertDataRow in insertData)
                {
                    ExcuteEntity re = new ExcuteEntity();
                    re.TableName = eachData["tableName"].ToString();
                    re.ExcuteType = ExcuteEntity.ExcuteTypeEnum.insert;
                    var excuteFileds = new List<FieldEntity>();
                    var uniqueColName = eachData["uniqueColName"].ToString().ToUpper();//唯一列
                    String uniqueColValue = "";//此行的唯一值
                    //获取唯一列的值
                    foreach (String key in insertDataRow.Keys) {
                        if (key.Split('@')[0].ToUpper() == uniqueColName)
                        {
                            uniqueColValue = insertDataRow[key].ToString();
                        }
                    }
                    //处理每列
                    foreach (String key in insertDataRow.Keys)
                    {
                        String[] _key = key.Split('@');
                        //获取目标数据库的映射字段
                        Dictionary<string, Object> toField = CommonHelper.getToField(thisTab, orgDb, toDb, _key[0]);
                        if (toField == null) {
                            continue;
                        }
                        if (toField["mapType"].ToString() == "str") {
                            _key[0] = toField[toDb].ToString();
                        }
                        if (toField["mapType"].ToString() == "attr")//此处要存入到属性表中
                        {
                            if (insertDataRow[key] != null && !string.IsNullOrWhiteSpace(insertDataRow[key].ToString()))
                            {
                                CommonHelper.saveAttrsData((JObject)toField[toDb], uniqueColValue, insertDataRow[key].ToString());
                                continue;
                            }
                        }
                        FieldEntity.FieldTypeEnum fieldType = FieldEntity.FieldTypeEnum.Str;
                        Object fieldVal = insertDataRow[key];
                        CommonHelper.GetFieldValAndType(_key, ref fieldType, ref fieldVal);
                        excuteFileds.Add(new FieldEntity(_key[0], fieldVal, fieldType));
                    }
                    re.ExcuteFileds = excuteFileds;

                    result.Add(re);
                }
                

            }
            return result;

        }
        #endregion

    }
}
